package io.github.wanggit.antrpc.client.config;

import io.github.wanggit.antrpc.client.zk.IZkClient;
import io.github.wanggit.antrpc.client.zk.ZkClient;
import io.github.wanggit.antrpc.client.zk.zknode.IZkNodeOperator;
import io.github.wanggit.antrpc.client.zk.zknode.ZkNodeOperator;
import io.github.wanggit.antrpc.commons.bean.ConfigItem;
import io.github.wanggit.antrpc.commons.bean.SubscribeConfigData;
import io.github.wanggit.antrpc.commons.config.ConfigChangeTypeEnum;
import io.github.wanggit.antrpc.commons.config.Configuration;
import io.github.wanggit.antrpc.commons.constants.ConstantValues;
import io.github.wanggit.antrpc.commons.utils.ConfigPathUtil;
import io.github.wanggit.antrpc.commons.utils.JsonUtils;
import io.github.wanggit.antrpc.commons.utils.NetUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.zookeeper.CreateMode;
import org.springframework.core.env.ConfigurableEnvironment;
import org.springframework.core.env.MutablePropertySources;
import org.springframework.core.env.PropertySource;
import org.springframework.util.StringUtils;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 *
 *
 * <pre>
 * 配置中心本地容器
 * /_rpc_cnf__/环境标示(默认default)/服务名或共享空间名/配置Key
 * /__rpc_cnf__/dev/service_a/config_name
 * </pre>
 */
@Slf4j
public class DefaultConfigContainer implements IConfigContainer {

    private final ConfigurableEnvironment environment;
    private final IConfigListener configListener;
    private static final Object lock = new Object();

    /** 一个Key有多个值时，取第一个。 ConfigItemTreeSet中的元素按优先级与创建时间从大到小排序, */
    private final ConcurrentHashMap<String, ConfigItemTreeSet> localStores =
            new ConcurrentHashMap<>();

    private final CopyOnWriteArrayList<IConfigChangeListener> configChangeListeners =
            new CopyOnWriteArrayList<>();

    public DefaultConfigContainer(ConfigurableEnvironment environment) {
        this.environment = environment;
        configListener = new DefaultConfigListener();
    }

    @Override
    public void addConfigChangeListener(IConfigChangeListener... listeners) {
        configChangeListeners.addAll(Arrays.asList(listeners));
    }

    @Override
    public void saveOrUpdate(ConfigItem configItem) {
        String key = configItem.getKey();
        synchronized (lock) {
            boolean needFireListener = !configChangeListeners.isEmpty();
            String oldValue = needFireListener ? getValue(key, null) : null;
            List<String> grayReleases = configItem.getGrayReleases();
            String hostInfo = getHostInfo();
            if (null != grayReleases
                    && !grayReleases.isEmpty()
                    && !grayReleases.contains(hostInfo)) {
                if (log.isInfoEnabled()) {
                    log.info(
                            "gray released list => "
                                    + JsonUtils.toJsonString(grayReleases)
                                    + ", hostInfo => "
                                    + hostInfo);
                }
                return;
            }
            if (!localStores.containsKey(key)) {
                localStores.put(key, new ConfigItemTreeSet());
            }
            calcConfigPriority(configItem);
            if (log.isDebugEnabled()) {
                log.debug("save or update " + JsonUtils.toJsonString(configItem));
            }
            ConfigItemTreeSet configItemTreeSet = localStores.get(key);
            configItemTreeSet.removeAndAddConfigItem(configItem);
            String newValue = configItemTreeSet.first().getValue();
            configItemToSpringEnvironment(key, newValue);
            if (needFireListener && !Objects.equals(oldValue, newValue)) {
                ConfigChangeTypeEnum changeTypeEnum =
                        null == oldValue ? ConfigChangeTypeEnum.ADD : ConfigChangeTypeEnum.UPDATE;
                configChangeListeners.forEach(
                        it -> it.changed(changeTypeEnum, configItem, oldValue, newValue));
            }
        }
    }

    @Override
    public void delete(ConfigItem configItem) {
        String key = configItem.getKey();
        calcConfigPriority(configItem);
        synchronized (lock) {
            if (localStores.containsKey(key)) {
                boolean needFireListener = !configChangeListeners.isEmpty();
                String oldValue = needFireListener ? getValue(key, null) : null;
                ConfigItemTreeSet configItemTreeSet = localStores.get(key);
                if (null != configItemTreeSet && !configItemTreeSet.isEmpty()) {
                    configItemTreeSet.removeConfigItem(configItem);
                }
                // 配置中心的所有配置都全部删除了，现在只有本地配置生效了
                if (null == configItemTreeSet || configItemTreeSet.isEmpty()) {
                    deleteConfigItemFromSpringEnvironment(key);
                    String maybeLocalValue = getValue(key, null);
                    if (needFireListener) {
                        configChangeListeners.forEach(
                                it ->
                                        it.changed(
                                                ConfigChangeTypeEnum.DELETE,
                                                configItem,
                                                oldValue,
                                                maybeLocalValue));
                    }
                } else if (!configItemTreeSet.isEmpty()) {
                    // 配置中心高等级配置删除，还有低等级的配置生效
                    String newEffectValue = localStores.get(key).first().getValue();
                    configItemToSpringEnvironment(key, newEffectValue);
                    if (needFireListener && !Objects.equals(oldValue, newEffectValue)) {
                        configChangeListeners.forEach(
                                it ->
                                        it.changed(
                                                ConfigChangeTypeEnum.DELETE,
                                                configItem,
                                                oldValue,
                                                newEffectValue));
                    }
                }
            }
        }
    }

    @Override
    public String getValue(String key, String defaultValue) {
        if (localStores.containsKey(key) && !localStores.get(key).isEmpty()) {
            return localStores.get(key).first().getValue();
        }
        return environment.getProperty(key, defaultValue);
    }

    @Override
    public void start() throws ConfigListenException {
        // 1. 计算要监听的节点路径
        List<String> paths = getAllPaths();
        if (log.isInfoEnabled()) {
            log.info("will be listen " + paths);
        }
        // 2. 先强制获取一次，再做监听节点，因为节点监听是异步的，所以需要进行一个同步获取的过程
        Configuration configuration = new Configuration();
        configuration.setZkIps(environment.getProperty(Constants.ZK_CONFIG, "localhost:2181"));
        IZkClient zkClient = new ZkClient(configuration);
        configListener.doFirstFetch(this, zkClient, paths.toArray(new String[] {}));
        configListener.listen(this, zkClient, paths.toArray(new String[] {}));
        // 3. 保存本节点订阅那些配置节点
        String hostInfo = getHostInfo();
        IZkNodeOperator zkNodeOperator = new ZkNodeOperator(zkClient.getCurator());
        for (String path : paths) {
            // /__rpc_subcnf__/env/spaceName/hostInfo
            String fullPath = ConfigPathUtil.castPathToSubscribePath(path) + "/" + hostInfo;
            if (log.isInfoEnabled()) {
                log.info("subscribe config path " + fullPath);
            }
            SubscribeConfigData subscribeConfigData = new SubscribeConfigData();
            subscribeConfigData.setPath(fullPath);
            subscribeConfigData.setTs(System.currentTimeMillis());
            zkNodeOperator.remoteCreateZkNode(
                    fullPath, JsonUtils.toJsonBytes(subscribeConfigData), CreateMode.EPHEMERAL);
        }
    }

    /**
     * 获取当前节点的 ip:port
     *
     * @return ip:port
     */
    public String getHostInfo() {
        return environment.getProperty(Constants.APPLICATION_EXPOSED_IP, NetUtil.getLocalIp())
                + ":"
                + environment.getRequiredProperty(Constants.APPLICATION_PORT);
    }

    /**
     * 获取应用的配置的Path
     *
     * @return application root path
     */
    private String getApplicationPath() {
        return ConfigPathUtil.getApplicationPath(
                environment.getProperty(
                        Constants.CONFIG_ENV_CONFIG, ConstantValues.CONFIG_DEFAULT_ENV),
                getApplicationName());
    }

    /**
     * 获取应用的配置的共享空间
     *
     * @param space space
     * @return shared space path
     */
    private String getSharedSpacePath(String space) {
        return ConfigPathUtil.getSharedSpacePath(
                environment.getProperty(
                        Constants.CONFIG_ENV_CONFIG, ConstantValues.CONFIG_DEFAULT_ENV),
                space.trim());
    }

    /**
     * 获取所有需要监听的路径集合
     *
     * @return paths
     */
    private List<String> getAllPaths() {
        List<String> paths = new ArrayList<>();
        paths.add(getApplicationPath());
        String sharedSpaces = environment.getProperty(Constants.SHARED_SPACES_CONFIG);
        if (StringUtils.hasText(sharedSpaces)) {
            String[] spaces = sharedSpaces.split(",");
            for (String space : spaces) {
                paths.add(getSharedSpacePath(space.trim()));
            }
        }
        return paths;
    }

    /**
     * 获取
     *
     * @return application name
     */
    private String getApplicationName() {
        return environment.getRequiredProperty(Constants.APPLICATION_NAME_CONFIG);
    }

    /**
     * 计算配置项的优化级
     *
     * @param configItem config item
     */
    private void calcConfigPriority(ConfigItem configItem) {
        // /__rpc_cnf__/default/service_a 包含应用名称就是本应用的配置，优化级最高
        if (configItem.getFullPath().contains(getApplicationName())) {
            configItem.setPriority(Constants.APPLICATION_CONFIG_LEVEL);
        } else {
            configItem.setPriority(Constants.SHARED_SPACE_CONFIG_LEVEL);
        }
    }

    /**
     * 从Spring 环境中删除对应的Key
     *
     * @param key key
     */
    private void deleteConfigItemFromSpringEnvironment(String key) {
        MutablePropertySources propertySources = environment.getPropertySources();
        if (log.isInfoEnabled()) {
            log.info("delete config from spring environment. key=" + key);
        }
        if (propertySources.contains(ConfigItemPropertySource.class.getSimpleName())) {
            PropertySource<?> propertySource =
                    propertySources.get(ConfigItemPropertySource.class.getSimpleName());
            if (null != propertySource) {
                ConfigItemPropertySource configItemPropertySource =
                        (ConfigItemPropertySource) propertySource;
                configItemPropertySource.removeProperty(key);
            }
        }
    }

    /**
     * 把获取到的配置添加到Spring环境中，如果已存在就替换。
     *
     * @param key config key
     * @param value config value
     */
    private void configItemToSpringEnvironment(String key, String value) {
        MutablePropertySources propertySources = environment.getPropertySources();
        if (log.isInfoEnabled()) {
            log.info("save config to spring environment. key=" + key);
        }
        String propertySourceName = ConfigItemPropertySource.class.getSimpleName();
        if (!propertySources.contains(propertySourceName)) {
            propertySources.addFirst(new ConfigItemPropertySource());
        }
        ConfigItemPropertySource configItemPropertySource =
                (ConfigItemPropertySource)
                        propertySources.get(ConfigItemPropertySource.class.getSimpleName());
        configItemPropertySource.setProperty(key, value);
    }
}
